package offline;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import redis.clients.jedis.Response;
import tlsb.Utils;
import util.IPartition;
import util.RedisCluster;

public class InvertedSetIndex implements ISetIndex {
	final private String PREFIXOFSID = "s:";
	final private String SETOFBELONGEDTL = ":blg";
	final private String SETOFFOLLOWERS = ":flr";

	RedisCluster iCluster;
	RedisCluster sCluster;
	IPartition iPart;
	IPartition sPart;
	CGAXiafan cga;

	/**
	 * 
	 * @param cluster_
	 *            存放计算后的sid的倒排信息的集群
	 * @param sCluster_
	 *            存放followship的集群
	 */
	public InvertedSetIndex(CGAXiafan cga_) {
		cga = cga_;
		iCluster = cga.getRetRedis();
		sCluster = cga.getSourceRedis();
		iPart = iCluster.getPartition();
		sPart = sCluster.getPartition();
	}

	@Override
	public boolean isExists(Set<String> set) {
		// TODO Auto-generated method stub
		return false;
	}

	private boolean switched = false;

	@Override
	public Map<String, Set<String>> findSubSetsMap(Set<String> set) {
		// 假设每个用户平均属于400个stl,一个用户的关注人数大概为400,那么最多需要返回的stl个数为160000，
		if (!switched) {

		}
		return findSubSetsMap_intern(set);
	}

	public Map<String, Set<String>> findSubSetsMap_intern(Set<String> set) {
		// 优化：对于那些uid的postList非常长的那种uid，不需要把它fetch过来，而是将它在已经fetch过来的sid中进行判断
		HashMap<String, Response<Set<String>>> postResMap = new HashMap<String, Response<Set<String>>>();
		int count = 0;
		for (String fle : set) {
			String fleCid = cga.getFirstCommunity(fle);
			postResMap.put(fle, iCluster.getPipeline(iPart.shard(fleCid))
					.smembers(Utils.joinStr(fle, SETOFBELONGEDTL)));
			if (count++ % 100 == 0)
				iCluster.synAllPipelines();
		}

		iCluster.synAllPipelines();

		Map<String, Set<String>> sidMap = new HashMap<String, Set<String>>();

		for (Entry<String, Response<Set<String>>> entry : postResMap.entrySet()) {
			Set<String> sids = entry.getValue().get();
			for (String sid : sids) {
				if (!sidMap.containsKey(sid)) {
					sidMap.put(sid, new HashSet<String>());
				}
				sidMap.get(sid).add(entry.getKey());
			}
		}

		for (Iterator<String> keyIter = sidMap.keySet().iterator(); keyIter
				.hasNext();) {
			String sid = keyIter.next();
			long card = sCluster.getJedis(sPart.shard((sid))).scard(
					Utils.joinStr(PREFIXOFSID, sid));
			if (card > sidMap.get(sid).size()) {
				keyIter.remove();
			}
		}
		return sidMap;
	}

	@Override
	public Set<String> findSubSets(Set<String> set) {
		return findSubSetsMap(set).keySet();
	}

	@Override
	public Set<String> findInterSets(Set<String> set) {

		return null;
	}

	@Override
	public void insertSet(Set<String> set, String setId) {
		// TODO Auto-generated method stub

	}

	@Override
	public Map<String, Set<String>> findInterSetMap(Set<String> set) {
		// 对于vip用于，由于其粉丝量非常大，因此不能将其对应的posterlistfetch过来，而是将其在和用户有关注交集的人进行probe
		// 缺点：丢失了vips可能产生的全局的交集

		HashMap<String, Response<Set<String>>> postResMap = new HashMap<String, Response<Set<String>>>();
		HashSet<String> vips = new HashSet<String>();

		for (String fle : set) {
			if (filter(fle)) {
				vips.add(fle);
			} else {
				postResMap.put(fle, sCluster.getPipeline(sPart.shard(fle))
						.smembers(Utils.joinStr(fle, SETOFFOLLOWERS)));
			}
		}

		sCluster.synAllPipelines();

		Map<String, Set<String>> uidToFlweeMap = new HashMap<String, Set<String>>();

		for (Entry<String, Response<Set<String>>> entry : postResMap.entrySet()) {
			Set<String> uids = entry.getValue().get();
			for (String uid : uids) {
				if (!uidToFlweeMap.containsKey(uid)) {
					uidToFlweeMap.put(uid, new HashSet<String>());
				}
				uidToFlweeMap.get(uid).add(entry.getKey());
			}
		}
		// TODO use pipeline
		for (String vip : vips) {
			HashMap<String, Response<Boolean>> cond = new HashMap<String, Response<Boolean>>();
			int count = 1;
			for (Entry<String, Set<String>> entry : uidToFlweeMap.entrySet()) {
				cond.put(entry.getKey(), isFollower(entry.getKey(), vip));

				if (count++ % 1000 == 0) {
					sCluster.synAllPipelines();
					for (Entry<String, Response<Boolean>> condEntry : cond
							.entrySet()) {
						if (condEntry.getValue().get()) {
							uidToFlweeMap.get(condEntry.getKey()).add(vip);
						}
					}
					cond.clear();
				}
			}
			if (cond.size() > 0) {
				sCluster.synAllPipelines();
				for (Entry<String, Response<Boolean>> condEntry : cond
						.entrySet()) {
					if (condEntry.getValue().get()) {
						uidToFlweeMap.get(condEntry.getKey()).add(vip);
					}
				}
			}
		}
		return uidToFlweeMap;
	}

	public Response<Boolean> isFollower(String flwer, String flwee) {
		return sCluster.getPipeline(sPart.shard(flwee)).sismember(
				Utils.joinStr(flwee, SETOFFOLLOWERS), flwer);
	}

	private boolean filter(String uid) {
		if (sCluster.getJedis(sPart.shard(uid)).scard(
				Utils.joinStr(uid, SETOFFOLLOWERS)) > 35000) {
			return true;
		} else {
			return false;
		}
	}
}
